#pragma once
#include"Any.hpp"
#include <iostream> 
#include <sys/eventfd.h>  
#include <sys/epoll.h>
#include <functional>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <cassert>
#include <unistd.h>
#include <sys/timerfd.h>
#include <unordered_map>
#include <memory>
#include"Log.hpp"
#include <cstring>
#include"Socket.hpp"
#include"../source/Log2.hpp"

// class Poller;
class EventLoop;
class Channel{
    private:
        int _fd;           //文件描述符,用于监控,
        EventLoop *_loop; //一个事件循环
        // Poller *_poller;  //一个事件监控 
        uint32_t _events; //当前监控的事件
        uint32_t _revents; //当前连接触发的事件

        using EventCallback = std::function<void()>;
        EventCallback _read_callback;  //可写事件回调
        EventCallback _write_callback; //可读
        EventCallback _error_callback;  //错误
        EventCallback _close_callback;  //连接断开
        EventCallback _event_callback;  //任意事件
    public:
        Channel(EventLoop* loop, int fd):_fd(fd),_events(0),_revents(0),_loop(loop){}
        int Fd() { return _fd; }    //获取文件描述符    
        uint32_t Events() { return _events; }  //获取监控的事件
        void SetREvents(uint32_t events) { _revents = events; }  //设置实际就绪的事件
        void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
        void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
        void SetErrorCallback(const EventCallback &cb) { _error_callback = cb ;}
        void SetCloseCallback(const EventCallback &cb) { _close_callback = cb ; }
        void SetEventCallback(const EventCallback &cb) { _event_callback = cb ;}
        //判断监控事件
        bool ReadAble() { return (_events & EPOLLIN); }
        bool WriteAble() { return (_events & EPOLLOUT); }
        
        void EnableRead() { _events |= EPOLLIN; Updata(); }
        void EnableWrite() { _events |= EPOLLOUT; Updata(); }
        void DisableRead() { _events &= ~EPOLLIN; Updata(); } 
        void DisableWrite() { _events &= ~EPOLLOUT; Updata(); }
        void DisableAll() { _events = 0; Updata(); }
        //移除监控
        void Remove();
        void Updata();
        
        void HandleEvent(){
            //可读，对端关闭，紧急事件可读。
            if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)){
                //任何事件都会触发的事件
                if(_read_callback) _read_callback();

            }else if(_revents & EPOLLOUT){
                if(_write_callback) _write_callback();
            }else if(_revents & EPOLLERR){
                if(_error_callback) _error_callback();
            }else if(_revents & EPOLLHUP){
                if(_close_callback) _close_callback();
            }
            if(_event_callback) _event_callback();
        }
    };

#define MAX_EPOLLEVENTS 1024
class Poller{
private:
    int _epfd;      //用一个文件描述符_epfd管理文件,
    struct epoll_event _evs[MAX_EPOLLEVENTS]; //事件队列
    std::unordered_map<int,Channel *> _channels;
private:
    //对epoll的直接操作 
    void Updata(Channel *channel ,int op){
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->Events();
        int ret = epoll_ctl(_epfd,op,fd,&ev); //系统调用
        if(ret<0){
            lg(Error,"EPOLLCTL FAILED!");//重复添加 ret<0
        }   
        return ;
    }
    //判断Channel是否添加了事件监控
    bool HashChannel(Channel *channel){
        auto it = _channels.find(channel->Fd());
        if(it == _channels.end()){
            return false;
        }
        return true;
    }
public:
    Poller(){
        _epfd = epoll_create(MAX_EPOLLEVENTS);  //系统调用创建epoll,MAX_EPOLLEVENTS只需要大于0,没有具体含义
        if(_epfd < 0){
            lg(Error,"EPOLL CREATE FAILED!!");
            abort(); //创建失败，退出
        }
    }
    //添加或修改监控事件
    void UpdataEvent(Channel *channel){
        bool ret = HashChannel(channel);
        if(ret == false){
            _channels.insert(std::make_pair(channel->Fd(),channel));
            return Updata(channel,EPOLL_CTL_ADD);
        }
        return Updata(channel,EPOLL_CTL_MOD);
    }
    //移除监控
    void RemoveEvent(Channel *channel){
        auto it = _channels.find(channel->Fd());
        if(it != _channels.end()){
            _channels.erase(it);
        }
        return Updata(channel,EPOLL_CTL_DEL);
    }
    //开始监控。
    void Poll(std::vector<Channel*> *active){
        int nfds = epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
        if(nfds < 0){
            if(errno == EINTR){ //EINTR：进程在
                //慢系统调用中阻塞，捕获到某个信号且相应的信号处理函
                return;
            }
            lg(Error,"EPOLL WAIT ERROR:%s\n",strerror(errno));
            abort();
        }
        for(int i=0;i < nfds;i++){//所有就绪的事件
            auto it = _channels.find(_evs[i].data.fd);
            assert(it !=  _channels.end());
            it ->second ->SetREvents(_evs[i].events);  //设置实际就绪的事件
            active->push_back(it->second);  //将就绪的事件放入active中
        }
        return;
    }    
};

//定时器模块
using TaskFunc = std::function<void()>;
using RealeaseFunc = std::function<void()>; 
class TimerTask{
private:
    uint64_t _id; //定时器id    
    uint32_t _timerout; //定时器超时时间    
    bool _canceled; // false-表示没有被取消，true-表示被取消    
    TaskFunc _task_cb; //定时器回调函数
    RealeaseFunc _release; //删除定时器
public:
    TimerTask(uint64_t id, uint32_t delay, const TaskFunc &cb):
        _id(id),_timerout(delay),_canceled(false),_task_cb(cb){} 
    ~TimerTask(){
        if(_canceled == false) _task_cb();
        _release();//删除定时器对象
    }
    void Cancel() { _canceled = true; }
    void setRelease (const RealeaseFunc &cb) { _release = cb;} 
    uint32_t DelayTime() { return _timerout; }
};

class TimerWheel{
private:
    using  WeakTask = std::weak_ptr<TimerTask>;
    using PtrTask = std::shared_ptr<TimerTask>;
    int _tick;  //当前的秒针
    int _capacity;  //表盘最大数量
    std::vector<std::vector<PtrTask>> _wheel;  
    std::unordered_map<uint64_t,WeakTask> _timers;
    
    EventLoop *_loop;
    int _timerfd;   //定时器描述符-可读事件回调就是读取计数器
    std::unique_ptr<Channel> _timer_channel;
private:    
    void RemoveTimer(uint64_t id){ //删除定时器 
        auto it = _timers.find(id);
        if(it != _timers.end()){
            _timers.erase(it);
        }
    }
    
    static int CreateTimerfd(){
        int timerfd = timerfd_create(CLOCK_MONOTONIC,0);//系统调用,设置定时事件，然后推动tick走
        if(timerfd < 0){
            lg(Error,"TIMERFD CREATE FAILED!");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;  
        itime.it_value.tv_nsec = 0; // 第一次超时事件为1s后 
        itime.it_interval.tv_sec = 1;       
        itime.it_interval.tv_nsec = 0;//第一次超时后,每次超时间隔事件
        timerfd_settime(timerfd,0,&itime, NULL);//设置时间
        return timerfd;
    }
    
    int ReadTimefd(){
        uint64_t times = 0;
        int ret = read(_timerfd, &times, 8);
        if(ret<0){
            lg(Error,"READ TIMEFD FAILED!");    
            abort();
        }
        return times;
    }

    void RunTimerTask(){
        _tick = (_tick + 1)%_capacity;
        _wheel[_tick].clear();//清空指定位置数据
    }
    void OnTime(){
        int times = ReadTimefd();
        for(int i=0;i<times;i++){ //超时次数，运行超时时间
            RunTimerTask();
        }
    }
    void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc &cb){
        PtrTask pt(new TimerTask(id,delay,cb)); //新的定时器任务
        pt->setRelease(std::bind(&TimerWheel::RemoveTimer,this,id));
        int pos = (_tick + delay) % _capacity;
        _wheel[pos].push_back(pt);
        _timers[id] = WeakTask(pt);
    }
    void TimerRefreshInLoop(uint64_t id){
        auto it = _timers.find(id);
        if(it == _timers.end()){
            return;
        }
        PtrTask pt = it->second.lock();//获取定时器
        int delay = pt->DelayTime();
        int pos = (delay + _tick) % _capacity;//刷新定时器
        _wheel[pos].push_back(pt);
    }

    void TimerCancelInLoop(uint64_t id){
        auto it = _timers.find(id);
        if(it == _timers.end()){
            return;
        }
        PtrTask pt = it->second.lock();
        if(pt){
            pt->Cancel();
        }  
    }
public:
    TimerWheel(EventLoop *loop):_capacity(60),_tick(0),_wheel(_capacity),_loop(loop),
        _timerfd(CreateTimerfd()),_timer_channel(new Channel(_loop,_timerfd)){
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
        _timer_channel->EnableRead();//包含Updata(),添加监控
    }
     /*如果不想加锁，那就把对定时器的所有操作，都放到⼀个线程中进⾏*/    
    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    // 接口存在线程安全问题。
    bool HashTimer(uint64_t id){
        auto it = _timers.find(id);
        if(it == _timers.end()){
            return false;
        }
        return true;
    }  
};

class EventLoop{
private:
    using Functor = std::function<void()>;
    std::thread::id _thread_id;//线程id
    int _event_fd;             
    std::unique_ptr<Channel> _event_channel;
    Poller _poller; //事件监控
    std::vector<Functor> _tasks;
    std::mutex _mutex;// 保证线程安全
    TimerWheel _timer_wheel;//定时器模块
public:
    //执行线程池中所有的任务
    void RunAllTask(){
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);//拿到线程池中的任务
        }
        for(auto &f : functor){
            f();
        }
        return;
    }


    static int CreateEventFd(){
        int efd = eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);//返回一个创建的文件描述符 efd,
        if(efd<0){
            lg(Error,"CREATE EVENTFD FAILED!!");
            abort();
        }
        return efd;
    }

    void ReadEventfd(){
        uint64_t res = 0;
        int ret = read(_event_fd,&res,sizeof(res));//res表示_event_fd写入值之和
        if(ret<0){
            if(errno == EINTR || errno == EAGAIN){ //  EAGAIN -- 
                //表⽰⽆数据可读
                return;
            }
            lg(Error,"READ EVENTFD FAILED!");
            abort();
        }
        return;
    }

    void WeakUpEventFd(){
        uint64_t val = 1;
        int ret = write(_event_fd,&val,sizeof(val));
        if(ret<0){
            if(errno == EINTR){
                return;
            }
            lg(Error,"READ EVENTFD FAILED!");
            abort();
        }
        return ;
    }
public:
    EventLoop():_thread_id(std::this_thread::get_id()),
                _event_fd(CreateEventFd()),_event_channel(new Channel(this,_event_fd))
                ,_timer_wheel(this) {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd,this));
        _event_channel->EnableRead();
    }
    
    void Start(){
        while(1){
            //1.事件监控
            std::vector<Channel *> actives;
            _poller.Poll(&actives);
            //2.事件处理
            for(auto &channel:actives){
                channel->HandleEvent();
                channel->Fd();
                // std::cout<<"channel->Fd()"<< channel->Fd()<<std::endl;
            }
            
            //3.执行任务
            RunAllTask();
        }
    }

    bool IsInLoop(){
        return  _thread_id == std::this_thread::get_id();
    }
    
    void AssertInLoop(){
        assert( _thread_id == std::this_thread::get_id());
    }
    
    void RunInLoop(const Functor &cb){
        if(IsInLoop()){
            return cb();
        }
        return QueueInLoop(cb);
    }
    
    void QueueInLoop(const Functor &cb){
        {   
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        //可能没有事件就绪，导致epoll阻塞
        //给eventfd写入一个数据，eventfd就会触发可读事件
        WeakUpEventFd();
    }
    
    void UpdataEvent(Channel *channel){
        return _poller.UpdataEvent(channel);
    }

    void RemoveEvent(Channel *channel){
        return _poller.RemoveEvent(channel);
    }

    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb){
        return _timer_wheel.TimerAdd(id,delay, cb);
    }
    void TimeRefresh(uint64_t id){
        return _timer_wheel.TimerRefresh(id);
    }
    void TimerCancel(uint64_t id){
        return _timer_wheel.TimerCancel(id);
    }
    bool HashTimer(uint64_t id){
        return _timer_wheel.HashTimer(id);
    } 
};

class LoopThread{
private:
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop *_loop;
    std::thread _thread; // EventLoop 对应的线程
private:
    void ThreadEntry(){
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        loop.Start();
    }
public:
    //创建线程，并关联loop
    LoopThread():_loop(NULL),_thread(std::thread(&LoopThread::ThreadEntry,this)){}
    EventLoop *GetLoop(){
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex); //加锁
            _cond.wait(lock, [&](){ return _loop !=NULL; });//条件不满足释放锁，并进入等待状态
            loop = _loop;
        }
        return loop;
    }
};

class LoopThreadPool{
private:
    int _thread_count; //线程数量
    int _next_idx;      //下一个线程
    EventLoop *_baseloop;   //基础线程
    std::vector<LoopThread*> _threads;  //线程池
    std::vector<EventLoop*> _loops;     //_loops
public:
    LoopThreadPool(EventLoop *baseloop):_thread_count(0),_next_idx(0)
        ,_baseloop(baseloop){}

    void SetThreadCount(int count) { _thread_count = count; }

    void Create(){
        if(_thread_count > 0){ //线程数量
            _threads.resize(_thread_count);//指针
            _loops.resize(_thread_count);
            for(int i=0; i<_thread_count; i++){
                _threads[i] = new LoopThread(); //创建线程
                _loops[i] = _threads[i]->GetLoop(); //关联线程中创建的loop
            }
        }
        return ; 
    }

    EventLoop *NextLoop(){
        if(_thread_count == 0){
            return _baseloop;
        }
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};

#define BUFFER_DEFAULT_SIZE 1024
class Buffer{
private: 
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;
public:
    Buffer():_reader_idx(0),_writer_idx(0),_buffer(BUFFER_DEFAULT_SIZE){}
    char *Begin(){ return &*_buffer.begin(); }
    char *WritePosition(){ return Begin()+_writer_idx; }
    char *ReadPosition() { return Begin() +_reader_idx; }
    uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }
    uint64_t HeadIdleSize() { return _reader_idx; }
    //可读大小
    uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }
    void MoveReadOffset(uint64_t len){ //读偏移
        if(len==0) return;
        assert(len <= ReadAbleSize());
        _reader_idx +=len;
    }
    void MoveWriteOffset(uint64_t len){ // 写偏移
        // std::cout<<"TailIdleSize():"<<TailIdleSize()<<std::endl;
        assert(len<=TailIdleSize());
        _writer_idx +=len;
    }
    //确保可写空间
    void EnsureWriteSpace(uint64_t len){
        if(TailIdleSize() >= len) { return; } 
        if(len <= TailIdleSize() + HeadIdleSize()){
            uint64_t rsz = ReadAbleSize();
            std::copy(ReadPosition(),ReadPosition()+rsz,Begin());//移动
            _reader_idx = 0;
            _writer_idx = rsz;
        }else{
            // std::cout<<"len:"<<len<<std::endl;
            lg(Info,"RESIZE %ld",_writer_idx+len);
            _buffer.resize(_writer_idx+len);
        }
    }
    //写入数据
    void Write(const void *data, uint64_t len){
        if(len == 0) return;
        EnsureWriteSpace(len);
        const char *d= (const char *) data;
        std::copy(d,d+len,WritePosition());
    }
    //
    void WriteAndPush(const void *data, uint64_t len){
        Write(data,len);
        MoveWriteOffset(len);
    }
    //
    void WriteString(const std::string &data){
        return Write(data.c_str(),data.size());
    }
    void WriteStringAndPush(const std::string &data){
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer &data) {
        return Write(data.ReadPosition(),data.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer &data){// connection联合调试，后来加的
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize());
    }
    void Read(void *buf, uint64_t len){
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(),ReadPosition()+len,(char*)buf);
    }
    void ReadAndPop(void *buf,uint64_t len){
        Read(buf,len);
        MoveReadOffset(len);
    }
    std::string ReadAsString(uint64_t len){
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0],len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len){
        assert(len <= ReadAbleSize());
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    char *FindCRLF(){
        char * res = (char*) memchr(ReadPosition(),'\n',ReadAbleSize());//序列化
        return res;
    }
    std::string GetLine(){
        char *pos = FindCRLF();
        if(pos == NULL){
            return "";
        }
        return ReadAsString(pos-ReadPosition()+1); //pos位置 减去ReadPosition()
    }
    std::string GetLineAndPop(){
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str; 
    }
    void clear(){
        _reader_idx =0;
        _writer_idx =0;
    }
};

class Connection;
typedef enum {DISCONNECTING, DISCONNECTED, CONNECTING, CONNECTED} ConnStatu;
using  PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection> {
private:
    //websocket
    // 添加协议类型枚举
    enum ProtocolType {
        TCP_PROTOCOL,
        WEBSOCKET_PROTOCOL
    };
    
    ProtocolType _protocol_type;  // 当前连接的协议类型
    
    // 添加 WebSocket 相关成员
    bool _is_websocket_handshake_done;
    
    // 添加 WebSocket 帧处理方法
    void handleWebSocketRead();
    void sendWebSocketMessage(const std::string& message);
    


    uint64_t _conn_id;  //连接id
    int _sockfd;    //关联文件描述符号
    bool _enable_inactive_release; //是否允许非活动释放
    EventLoop *_loop;   //连接关联的事件循环
    ConnStatu _statu;   //连接状态
    Socket _socket;     //套接字操作管理
    Channel _channel;   //连接的事件管理
    Buffer _in_buffer;   //输入缓冲区，在水平触发模式下，只要发送缓冲区(内核)有空间可写，epoll_wait 就会持续通知写事件。
    Buffer _out_buffer; //输出缓冲区
    Any _context;       //请求的接收处理上下文

    //四个回调函数
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;  
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    ConnectedCallback _connected_callback;    
    MessageCallback _message_callback;
    ClosedCallback _close_callback;
    AnyEventCallback _event_callback;
    ClosedCallback _server_closed_callback;
private:
    /* 五个channel的事件回调函数*/
    void HandleRead(){
        char buf[65536];
        ssize_t ret = _socket.NonBlockRecv(buf,65535); 
        if(ret<0){ //关闭
            return ShutdownInLoop();  
        }
        //数据放进输入缓冲区,
        _in_buffer.WriteAndPush(buf,ret);
        if(_in_buffer.ReadAbleSize() > 0){ //读取数据
            return _message_callback(shared_from_this(), &_in_buffer); //处理数据,读取数据到inbuffer里
        }
    }
    void HandleWrite(){ //写事件,发送完buffer里的数据就直接关闭连接
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
        if(ret<0){
            //发送错误就该关闭连接了
            if(_in_buffer.ReadAbleSize() > 0){ //先处理缓存区的读取数据
                _message_callback(shared_from_this(),&_in_buffer);  
            }
            return Release(); //这时候就是实际的关闭释放操作了。
        }
        
        _out_buffer.MoveReadOffset(ret);    
        if(_out_buffer.ReadAbleSize() == 0){
            _channel.DisableWrite(); //关闭写事件,水平则一直触发
            if(_statu == DISCONNECTING){ //关闭连接
                return Release();
            }
        }
    } 
    void HandleClose(){
        if(_in_buffer.ReadAbleSize() > 0){
            _message_callback(shared_from_this(),&_in_buffer);
        }
        return Release();
    }
    void HandleError(){
        return HandleClose();
    }
    void HandleEvent(){
        if(_enable_inactive_release == true){
            _loop->TimeRefresh(_conn_id); //默认事件刷新timer,如果channel每个处理之前都设置event_handle,刷新在事件处理之前,不会导致connection release。
        }
        if(_event_callback) _event_callback(shared_from_this());
    }
    //连接建立,状态设置
    void EstablishedInLoop(){
        assert(_statu == CONNECTING);
        _statu = CONNECTED;
        _channel.EnableRead();  
        if(_connected_callback) _connected_callback(shared_from_this());
    }
    //释放的实际接口
    void ReleaseInLoop(){
        _statu = DISCONNECTED;
        //移除事件监控
        _channel.Remove();
        _socket.Close();    
        if(_loop->HashTimer(_conn_id)) CancelInactiveReleaseInLoop();
        if(_close_callback) _close_callback(shared_from_this());
        if(_server_closed_callback) _server_closed_callback(shared_from_this());
    }
    //把数据发送到缓冲区
    void SendInLoop(Buffer &buf){
        if(_statu  == DISCONNECTED){
            return;
        }   
        _out_buffer.WriteBufferAndPush(buf); // 发送数据
        if(_channel.WriteAble()==false){
            _channel.EnableWrite(); 
        }
    }
    // 连接释放操作，需要判断还有没有数据待处理
    void ShutdownInLoop(){
        _statu = DISCONNECTING;
        if(_in_buffer.ReadAbleSize() > 0){
            if(_message_callback) _message_callback(shared_from_this(),&_in_buffer);
        }
        if(_out_buffer.ReadAbleSize() > 0){
            if(_channel.WriteAble() == false){
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize() == 0){
            Release();
        }
    }
    //启动非活跃释放超时规则
    void EnableInactiveReleaseInLoop(int sec){ 
        _enable_inactive_release = true;
        if(_loop->HashTimer(_conn_id)){
            return _loop->TimeRefresh(_conn_id);
        }
        _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::Release,this)); //定时器添加
    }
        
    //关闭非活跃释放超时规则
    void CancelInactiveReleaseInLoop(){
        _enable_inactive_release = false;
        if(_loop->HashTimer(_conn_id)){
            _loop->TimerCancel(_conn_id);
        }
    }
    
    //更新四个回调函数,
    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn,
        const MessageCallback &msg, const ClosedCallback &close
        ,const AnyEventCallback &event){
        _context = context;
        _connected_callback = conn; 
        _message_callback = msg;
        _close_callback = close;    
        _event_callback = event;
    }
public://对外接口

    //websocket
    // 添加协议升级方法
    void upgradeToWebSocket();
    
    // 重载 Send 方法以支持 WebSocket
    void Send(const char* data, size_t len);

    Connection(EventLoop *loop,uint64_t conn_id,int sockfd):_conn_id(conn_id),_sockfd(sockfd)
            ,_enable_inactive_release(false),_loop(loop),_statu(CONNECTING),_socket(_sockfd)
            ,_channel(loop,_sockfd){
                _channel.SetReadCallback(std::bind(&Connection::HandleRead,this)); 
                _channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
                _channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
                _channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
                _channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));    
                
            }
    ~Connection() { lg(Info,"Connection Release!");  }
    int Fd(){
        return _sockfd;
    }
    int Id(){
        return _conn_id;
    }
    //是否处于CONNECTED状态
    bool Connected(){
        return   (_statu == CONNECTED); 
    }   
    void SetContext(const Any &context ){ _context = context; }  
    //获取连接的上下文
    Any *GetContext(){ return &_context; }
    void SetConectedCallback(const ConnectedCallback &cb){
         _connected_callback = cb; 
    }
    void SetMessageCallback(const MessageCallback &cb){
        _message_callback = cb;
    }
    void SetCloseCallback(const ClosedCallback &cb){
        _close_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb){
        _event_callback = cb;
    }
    void SetSvrConectedCallback(const ConnectedCallback &cb){
        _server_closed_callback = cb; 
   }
    //连接建立,设置channel回调设置
    void Established(){
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }
    //
    void Send(const char* data, size_t len){
        Buffer buf;
        buf.WriteAndPush(data,len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf)));
    }
    void Shutdown(){
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
    }
    void Release(){
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
    }
    void EnabelInactiveRelease(int sec){
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));  
    }
    void CacelInactiveRelease(){
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
    }
    //切换协议，重置上下文以及阶段性回调函数 -- 而是这个接口必须在eventloop
    //线程中立即执行，防备新的事件触发后,处理的时候，切换任务还没有被执行
    //--会导致数据使用原协议了
    void Upgrade(const Any &context, const ConnectedCallback &conn
        , const MessageCallback &msg, const ClosedCallback &close
        , const AnyEventCallback &event){
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,close,event));
    }
    
};

class Acceptor{
private:
    Socket _socket;
    EventLoop *_loop;
    Channel _channel;
    
    using AcceptCallback = std::function<void(int)> ;
    AcceptCallback _accept_callback;
private:
    void HandleRead(){ //设置给acceptor接受建立的_channel的回调函数
        int newfd = _socket.Accept(); //监听套接字监听并建立连接
        if(newfd < 0){
            return ;
        }
        if(_accept_callback) _accept_callback(newfd);//
    }
    void HandleEvent(){    
        std::cout<<"_channel->Fd():"<<_channel.Fd()<<std::endl;
    }
    int CreateServer(int port){ 
        bool ret = _socket.CreateServer(port);
        assert(ret == true);
        return _socket.Fd();
    }
public:
    Acceptor(EventLoop *loop,int port):_socket(CreateServer(port)),_loop(loop)
        ,_channel(loop,_socket.Fd()){   //建立连接
            _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
            _channel.SetEventCallback(std::bind(&Acceptor::HandleEvent, this));
    }
    ~Acceptor(){}
    void SetAcceptCallback(const AcceptCallback &cb){ //设置回调函数
        _accept_callback = cb;
    }
    void Listen(){ //事件管理挂载
        _channel.EnableRead();
    }
};

class TcpServer{
private:
    uint64_t _next_id;
    int _port;
    int _timeout;
    bool _enable_inactive_release;
    EventLoop _baseloop;
    Acceptor _acceptor;     //监听套接字管理对象
    LoopThreadPool _pool;   //这是从属EventLoop线程池
    //保存管理所有连接对应的shared_ptr对象
    std::unordered_map<uint64_t,PtrConnection> _conns;
    
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    using Functor = std::function<void()>;
    //4个回调函数
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
private:
    void RunAfterInLoop(const Functor &task, int delay){
        _next_id++;
        _baseloop.TimerAdd(_next_id,delay,task);
    }

    void NewConnection(int fd){
        _next_id++;
        PtrConnection conn(new Connection(_pool.NextLoop(),_next_id,fd));
        conn->SetMessageCallback(_message_callback);
        conn->SetCloseCallback(_closed_callback);
        conn->SetConectedCallback(_connected_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetSvrConectedCallback(std::bind \
            (&TcpServer::RemoveConnection,this,std::placeholders::_1));
        if(_enable_inactive_release) conn->EnabelInactiveRelease(_timeout);
        conn->Established();
        _conns.insert(std::make_pair(_next_id,conn));
    }
    void RemoveConnection(const PtrConnection &conn){//线程安全,确保EventLoop对象\
        在自己的thread_id 线程中运行。如果是从属线程中调用RemoveConnection的话,\
        _baseloop的线程id和运行的线程id不一致。
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }
    void RemoveConnectionInLoop(const PtrConnection &conn){
        int id = conn->Id();
        auto it = _conns.find(id);
        if( it != _conns.end()){ //删除conn
            _conns.erase(it);
        }
    }
public:
    TcpServer(int port):
        _port(port),
        _next_id(0),
        _enable_inactive_release(false),
        _acceptor(&_baseloop,port),
        _pool(&_baseloop)
        {
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, \
                this, std::placeholders::_1));
            _acceptor.Listen();
        }
    void SetThreadCount(int count){
        return _pool.SetThreadCount(count);
    }
    void SetCounnectedCallback(const ConnectedCallback &cb){
        _connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback &cb){
        _message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback &cb){
        _closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback &cb){
        _event_callback = cb;
    }
    void EnableInactiveRelease(int timeout){
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    void RunAfter(const Functor &task, int delay){
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
    }
    void Start(){ 
        _pool.Create();
        _baseloop.Start();
    }
};

void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc &cb){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));       
}

void TimerWheel::TimerRefresh(uint64_t id){
    // DBG_LOG("TimerRefreshrun");
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));       
}

void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));    
}


void Channel::Remove() {  _loop->RemoveEvent(this); } 
void Channel::Updata() {  _loop->UpdataEvent(this); }


// 在 Connection 类中添加以下方法：

void Connection::upgradeToWebSocket() {
    _protocol_type = WEBSOCKET_PROTOCOL;
    _is_websocket_handshake_done = true;
    lg.info("Connection upgraded to WebSocket protocol");
}

void Connection::Send(const char* data, size_t len) {
    if (_protocol_type == WEBSOCKET_PROTOCOL && _is_websocket_handshake_done) {
        // 使用 WebSocket 协议发送
        std::string frame = WebSocket::encodeFrame(std::string(data, len));
        Buffer buf;
        buf.WriteAndPush(frame.c_str(), frame.length());
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    } else {
        // 使用原始 TCP 协议发送
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }
}

void Connection::handleWebSocketRead() {
    // 处理 WebSocket 帧
    while (_in_buffer.ReadAbleSize() > 0) {
        std::string message;
        WebSocket::Opcode opcode;
        
        // 尝试解码一个 WebSocket 帧
        std::string bufferData = _in_buffer.ReadAsString(_in_buffer.ReadAbleSize());
        if (WebSocket::decodeFrame(bufferData.c_str(), bufferData.length(), message, opcode)) {
            // 处理不同类型的 WebSocket 帧
            switch (opcode) {
                case WebSocket::TEXT:
                case WebSocket::BINARY:
                    // 将消息传递给业务处理器
                    if (_message_callback) {
                        // 创建一个临时缓冲区来传递消息
                        Buffer tempBuffer;
                        tempBuffer.WriteAndPush(message.c_str(), message.length());
                        _message_callback(shared_from_this(), &tempBuffer);
                    }
                    break;
                    
                case WebSocket::PING:
                    // 回复 PONG
                    sendWebSocketMessage("");
                    break;
                    
                case WebSocket::PONG:
                    // 忽略 PONG
                    break;
                    
                case WebSocket::CLOSE:
                    // 关闭连接
                    Shutdown();
                    return;
                    
                default:
                    break;
            }
            
            // 移除已处理的数据
            _in_buffer.MoveReadOffset(bufferData.length());
        } else {
            // 数据不完整，等待更多数据
            break;
        }
    }
}